package org.hyf.inspur.LessonDesin.clear.count4;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ProMain {
    public static void main(String[] args) throws Exception{
        //System.setProperty("hadoop.home.dir", "D://hadoop//hadoop1//hadoop-common-2.2.0-bin-master");
        try {
            //创建配置信息
            Configuration conf = new Configuration();
            //map内存设置
            conf.set("mapreduce.map.memory.mb", "5120");
            conf.set("mapreduce.reduce.memory.mb", "5120");
            // 创建任务
            Job job = new Job(conf, "ProMain");
            // 打成jar包运行，这句话是关键
            job.setJarByClass(ProMain.class);
            // 设置自定义Mapper类和设置map函数输出数据的key和value的类型
            job.setMapperClass(ProMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            //分区函数
            job.setPartitionerClass(KeyPartitioner.class);
            //分组函数
            job.setGroupingComparatorClass(KeyGroupingComparator.class);
            //  设置reduce数量为0
            job.setNumReduceTasks(1);
            job.setReducerClass(ProReducer.class);
            //结果数据输出类型
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Text.class);
            // 1.1 设置输入目录和设置输入数据格式化的类
            FileInputFormat.addInputPath(job, new Path("D://data/output/output1/part-m-00*"));
            FileInputFormat.addInputPath(job, new Path("D://data/input/configData/t_dx_product_msg_addr.txt"));
            //FileInputFormat.setInputPaths(job, "file:///D:/All-Guo/DailyStudy/DataProject/data/02.DxFileMatch/part-m-00*");
            //job.addCacheFile(new Path("file:///D:/All-Guo/DailyStudy/DataProject/配置数据/t_dx_product_msg_addr.txt").toUri());
            FileOutputFormat.setOutputPath(job, new Path("D://data/output/output5"));
            //提交作业 判断退出条件（0正常退出，1非正常退出）
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    //更改分区函数类 用key中的行为id作为分区数据
    public static class KeyPartitioner extends Partitioner<Text, Text> {
        @Override
        public int getPartition(Text key, Text value, int numPartitions) {
            return Math.abs(key.toString().split(",")[0].hashCode() * 127) % numPartitions;
        }
    }
    //更改分区函数类 相同分组数据分配到同一个reducer
    public static class KeyGroupingComparator extends WritableComparator {
        protected KeyGroupingComparator() {
            super(Text.class, true);
        }
        //比较判断
        @Override
        public int compare(WritableComparable w1, WritableComparable w2) {
            Text ip1 = (Text) w1;
            Text ip2 = (Text) w2;
            String l = ip1.toString().split(",")[0];
            String r = ip2.toString().split(",")[0];
            return l.equals(r) ? 0 : 1;
        }
    }
}
